#!/bin/bash
# Kafka 集群管理脚本
# 用法: ./kafka-cluster.sh [start|stop|status|restart]

KAFKA_HOME="/opt/kafka/kafka"
declare -a configs=(
  "server1.properties"
  "server2.properties"
  "server3.properties"
)

case "$1" in
  start)
    echo "启动 Kafka 集群..."
    for config in "${configs[@]}"; do
      echo "启动 broker $config..."
      nohup "$KAFKA_HOME/bin/kafka-server-start.sh" "$KAFKA_HOME/config/$config" > "$KAFKA_HOME/tmp/kafka-$config.log" 2>&1 &
      sleep 2
    done
    ;;

  stop)
    echo "停止 Kafka 集群..."
    "$KAFKA_HOME/bin/kafka-server-stop.sh"
    # 确保所有进程停止
    sleep 5
    pkill -f "kafka.Kafka" || true
    ;;

  status)
    echo "Kafka 集群状态:"
    for config in "${configs[@]}"; do
      broker_id=$(grep "broker.id" "$KAFKA_HOME/config/$config" | cut -d'=' -f2)
      port=$(grep "listeners=PLAINTEXT://:" "$KAFKA_HOME/config/$config" | cut -d':' -f4)

      if netstat -tuln | grep ":$port" > /dev/null; then
        echo "  Broker $broker_id (端口 $port): 运行中"
      else
        echo "  Broker $broker_id (端口 $port): 未运行"
      fi
    done
    ;;

  restart)
    $0 stop
    sleep 5
    $0 start
    ;;

  *)
    echo "用法: $0 {start|stop|status|restart}"
    exit 1
esac

exit 0